package com.yanjin.rxjavademo;

import android.support.annotation.NonNull;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.text.TextUtils;
import android.util.Log;
import android.view.View;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.schedulers.Schedulers;


public class MainActivity extends AppCompatActivity {
    private Disposable mDisposable;
    private CompositeDisposable mCompositeDisposable;
    private int y = 0;
    private Subscription mSubscription = null;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        findViewById(R.id.btn_rxjava_one).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                test();
            }
        });
        findViewById(R.id.btn_rxjava_tow).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                testTow();
            }
        });
        findViewById(R.id.btn_rxjava_tri).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                testTri();
            }
        });
        findViewById(R.id.btn_rxjava_for).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                testFor();
            }
        });
        findViewById(R.id.btn_rxjava_five).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                testFive();
            }
        });
        findViewById(R.id.btn_rxjava_six).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                testSix();
            }
        });
        findViewById(R.id.btn_rxjava_seven).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                testSeven();
            }
        });
        mCompositeDisposable = new CompositeDisposable();
    }

    /**
     * 入门，了解上游与下游的发射顺序
     */
    public void test() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) {
                //上游可以发送无限个onNext, 下游也可以接收无限个onNext.
                //当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送, 而下游收到onComplete事件之后将不再继续接收事件.
                //当上游发送了一个onError后,  上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件.
                //上游可以不发送onComplete或onError.
                //最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError,  也不能先发一个onComplete, 然后再发一个onError, 反之亦然
                //但若是发送多个onError, 则收到第二个onError事件会导致程序会崩溃.

                e.onNext("11111");
                Log.d("yanjin", "发送一个数据---11111----");
                e.onNext("22222");
                Log.d("yanjin", "发送一个数据---22222----");
//                e.onError(new Throwable("测试错误"));
//                Log.d("yanjin", "测试错误-------");
                e.onNext("33333");
                Log.d("yanjin", "发送一个数据---33333----");
                e.onComplete();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                //Disposable是暂停器，调用他的disposable方法是可以暂停接收的数据的
                //上游是不会停止发送的，只是下游不再接收。
                mDisposable = d;
                if (mCompositeDisposable != null) {
                    mCompositeDisposable.add(mDisposable);
                }
                Log.d("yanjin", "onSubscribe-------");
            }

            @Override
            public void onNext(String value) {
                //注意: 调用dispose()并不会导致上游不再继续发送事件, 上游会继续发送剩余的事件.只是下游不再接收。
                mDisposable.dispose();
                Log.d("yanjin", "接收的信息为-----" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d("yanjin", "onError-----");
            }

            @Override
            public void onComplete() {
                Log.d("yanjin", "onComplete-------");
            }
        });
    }

    /**
     * Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
     * Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
     * Schedulers.newThread() 代表一个常规的新线程
     * AndroidSchedulers.mainThread()  代表Android的主线程
     */
    private void testTow() {
        y = 0;
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                for (int i = 0; i < 500000000; i++) {
                    y = y + i;
                }
                e.onNext(y + "");
                e.onComplete();
            }
        }).subscribeOn(Schedulers.newThread())//多次指定上游的线程只有第一次指定的有效, 也就是说多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略.
                .observeOn(AndroidSchedulers.mainThread())//多次指定下游的线程是可以的, 也就是说每调用一次observeOn() , 下游的线程就会切换一次.
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        //那如果有多个Disposable 该怎么办呢, RxJava中已经内置了一个容器CompositeDisposable,
                        //每当我们得到一个Disposable时就调用CompositeDisposable.add()将它添加到容器中,
                        //在退出的时候, 调用CompositeDisposable.clear() 即可切断所有的水管.
                        mCompositeDisposable.add(d);
                    }

                    @Override
                    public void onNext(String value) {
                        Log.d("yanjin", "当前计算结果为----" + value);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    /**
     * map是RxJava中最简单的一个变换操作符了, 它的作用就是对上游发送的每一个事件应用一个函数, 使得每一个事件都按照指定的函数去变化.
     * 比如将上游的int修改为String给下游
     */
    public void testTri() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onNext(4);
                e.onComplete();
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return integer + "";
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String value) {
                Log.d("yanjin", "onNext-----" + value);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                Log.d("yanjin", "onComplete-----");
            }
        });
    }

    /**
     * flatMap是将上游的onNext方法转换成多个，例子中我将它增加为两个
     * Observable.fromIterable是可以加延迟的，那么加上后可能出现顺序错乱，这个时候就要用concatMap
     */
    private void testFor() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onNext(4);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                List<String> list = new ArrayList<>();
                list.add(integer + "复制一");
                list.add(integer + "复制二");
                return Observable.fromIterable(list).debounce(10, TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d("yanjin", "flatMap转换后输出的当前值==" + s);
            }
        });
    }

    /**
     * zip是将两个发射器的内容结合，当他们都在主线程的时候，会发现log是先显示发射器1在显示发射器2
     * 原因是两根水管都是运行在同一个线程里, 同一个线程里执行代码肯定有先后顺序
     * 解决这个问题就要他们在不同线程就行。
     * <p>
     * 这个操作符适合用来同时请求或查询两个数据，最后整合在一起的情况。
     */
    private void testFive() {
        Observable<String> observable1 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("A");
                Log.d("yanjin", "一号发射器---A-");
                e.onNext("B");
                Log.d("yanjin", "一号发射器---B-");
                e.onNext("C");
                Log.d("yanjin", "一号发射器---C-");
                e.onNext("D");
                Log.d("yanjin", "一号发射器---D-");
                e.onComplete();
                Log.d("yanjin", "一号发射器---onComplete-");
            }
        }).subscribeOn(Schedulers.io());
        Observable<Integer> observable2 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                Log.d("yanjin", "二号发射器---1-");
                e.onNext(2);
                Log.d("yanjin", "二号发射器---2-");
                e.onNext(3);
                Log.d("yanjin", "二号发射器---3-");
                e.onNext(4);
                Log.d("yanjin", "二号发射器---4-");
                e.onNext(5);
                Log.d("yanjin", "二号发射器---5-");
                e.onComplete();
                Log.d("yanjin", "一号发射器---onComplete-");
            }
        }).subscribeOn(Schedulers.io());
        Observable.zip(observable1, observable2, new BiFunction<String, Integer, String>() {
            @Override
            public String apply(String s, Integer integer) throws Exception {
                return s + integer;
            }
        }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String value) {
                        Log.d("yanjin", "onNext-----" + value);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        Log.d("yanjin", "onComplete-----");
                    }
                });
    }

    /**
     * zip会有一个水缸存储每一个发射器没来得及发射出去的事件，这个水缸的存储规则也是先进先出，
     * 但是如果一个发射器在不停的发送信息，即使下游没有接受，但是会存到水缸里，内存也会不断增加
     * 原因是他们在不同的线程，异步关系不会关心其他人只会关心自己的发送，发了一个，再来一个
     * 如果都在同一个线程就没有这个情况，因为同一个线程都会有先后顺序。
     * <p>
     * 其实解决这样的问题，我们可以在数量上和速度上解决问题，数量上我们隔几秒抽取一个数据，速度上，我们睡两秒发送一个数据
     */
    private void testSix() {
        Observable<String> observable1 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("A");
            }
        }).subscribeOn(Schedulers.io());

        Observable<Integer> observable2 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                for (int x = 0; ; x++) {
                    e.onNext(x);
                }
            }
        }).subscribeOn(Schedulers.io());
        Observable.zip(observable1, observable2, new BiFunction<String, Integer, String>() {
            @Override
            public String apply(String s, Integer integer) throws Exception {
                return s + integer;
            }
        }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String value) {
                        Log.d("yanjin", "onNext-------" + value);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }


    /**
     * 当上下游工作在不同的线程里时，每一个线程里都有一个requested，而我们调用request（1000）时，实际上改变的是下游主线程中的requested，
     * 而上游中的requested的值是由RxJava内部调用request(n)去设置的，这个调用会在合适的时候自动触发。
     */
    private void testSeven() {
        Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> e) throws Exception {
                for (int x = 0; ; x++) {
                    if(e.isCancelled()){
                        break;
                    }
                    if(e.requested()==0){
                        break;
                    }
                    e.onNext(x + "");
                    Log.d("yanjin", "已发送=" + e.requested());
                }
            }
        }, BackpressureStrategy.ERROR).map(new Function<String, String>() {
            @Override
            public String apply(String s) {
                return "接收到发来的数据" + s;
            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        mSubscription = s;
                        mSubscription.request(200);
                    }

                    @Override
                    public void onNext(String s) {
                        Log.d("yanjin", "next===" + s);
                        mSubscription.request(200);
                    }

                    @Override
                    public void onError(Throwable t) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        mCompositeDisposable.clear();
        mSubscription.cancel();
    }
}
